//
// Created by 49647 on 2022/4/10.
//

#include "ALooper.h"
#include <chrono>
#include "AHandler.h"
#include "AMessage.h"
#include "log.h"

#define LOG_TAG "ALooper"
using namespace std;

/* static */
int64_t ALooper::GetNowUs() {
    chrono::system_clock::duration t = chrono::system_clock::now().time_since_epoch();
    return chrono::duration_cast<chrono::microseconds>(t).count();
}

std::shared_ptr<ALooper> ALooper::Create() {
    return std::shared_ptr<ALooper>(new ALooper(), Destroy);
}

void ALooper::Destroy(ALooper *looper) {
    delete looper;
}

ALooper::~ALooper() {
    LOGI(">>");
    stop();
}

void ALooper::setName(const std::string &name) {
    mName = name;
}

void ALooper::registerHandler(const shared_ptr<AHandler> &handler) {
    lock_guard<mutex> lck(mLock);
    mHandler = handler;
    handler->setLooper(weak_from_this());
}

void ALooper::unregisterHandler() {
    lock_guard<mutex> lck(mLock);
    mHandler.reset();
}

int ALooper::start() {
    LOGI(">>");
    lock_guard<mutex> lck(mLock);
    if (mThread != nullptr) {
        LOGE("thread already created");
        return -ENOSYS;
    }
    mThread = make_unique<thread>([this] {
        do {
        } while (loop());
    });
    return 0;
}

void ALooper::stop() {
    LOGI(">>");
    mExit = true;
    mQueueChangedCondition.notify_one();
    {
        lock_guard<mutex> lck(mLock);
        if (mThread != nullptr and mThread->joinable()) {
            mThread->join();
            mThread.reset();
        }
    }
    LOGI("thread joined");
    {
        lock_guard<mutex> autoLock(mRepliesLock);
        mRepliesCondition.notify_all();
    }
}

void ALooper::post(const shared_ptr<AMessage> &msg, int64_t delayUs) {
    lock_guard<mutex> lck(mLock);

    int64_t whenUs;
    if (delayUs > 0) {
        int64_t nowUs = GetNowUs();
        whenUs = (delayUs > INT64_MAX - nowUs ? INT64_MAX : nowUs + delayUs);
    } else {
        whenUs = GetNowUs();
    }

    auto it = mEventQueue.begin();
    while (it != mEventQueue.end() && it->mWhenUs <= whenUs) {
        ++it;
    }
    if (it == mEventQueue.begin()) {
        mQueueChangedCondition.notify_one();
    }

    mEventQueue.insert(it, Event{.mWhenUs = whenUs, .mMessage = msg});
}

bool ALooper::loop() {
    Event event;
    {
        unique_lock<mutex> lck(mLock);
        if (mThread == nullptr or mExit) {
            return false;
        }
        if (mEventQueue.empty()) {
            LOGI("queue empty, wait");
            mQueueChangedCondition.wait(lck);
            LOGI("wake up");
            return true;
        }
        int64_t whenUs = mEventQueue.begin()->mWhenUs;
        int64_t nowUs = GetNowUs();

        if (whenUs > nowUs) {
            int64_t delayUs = whenUs - nowUs;
            if (delayUs > INT64_MAX / 1000) {
                delayUs = INT64_MAX / 1000;
            }
            mQueueChangedCondition.wait_for(lck, chrono::microseconds(delayUs));
            return true;
        }

        event = *mEventQueue.begin();
        mEventQueue.erase(mEventQueue.begin());
    }

    event.mMessage->deliver();
    return true;
}

shared_ptr<AReplyToken> ALooper::createReplyToken() {
    return shared_ptr<AReplyToken>(new AReplyToken(shared_from_this()));
}

int ALooper::awaitResponse(const shared_ptr<AReplyToken> &replyToken, shared_ptr<AMessage> *response) {
    unique_lock<mutex> lck(mRepliesLock);
    while (!replyToken->retrieveReply(response)) {
        {
            lock_guard<mutex> lock(mLock);
            if (mThread == nullptr) {
                return -ENOENT;
            }
        }
        mRepliesCondition.wait(lck);
    }
    return 0;
}

int ALooper::postReply(const shared_ptr<AReplyToken> &replyToken, const shared_ptr<AMessage> &reply) {
    lock_guard<mutex> lck(mRepliesLock);
    int err = replyToken->setReply(reply);
    if (err == 0) {
        mRepliesCondition.notify_all();
    }
    return err;
}


